from kafka import KafkaConsumer
import time
import json
import queue
import sumo_main
from constants import KAFKA_HOST, VIR_DEVICE_STATUS_TOPIC, AV_VEHICLE_TOPIC,MTF_VIR_DEVICE_STATUS_TOPIC
real_vehicle_queue = queue.Queue(maxsize=2000)

def receive_message(userId, task_id, resource):
    print('kafka 消费被触发')
    # 首先定义全局变量，用于在不同方法内互相通信
    global vir_gantry_dict, vir_ramp_light, SPEED_LIMIT_SIGN, \
        speed_limit_point_list, speed_limit_list, speed_gantry_asc, \
        emergency_dict, special_dict
    key = None
    if resource == 0:
        # key = str(userId) + '_' + str(task_id) + '_' + str(resource)
        key = str(task_id)
    elif resource == 1:
        key = str(task_id)

    # key = str(579) + '_' + str(224) + '_' + str(1)
    consumer = KafkaConsumer(
        MTF_VIR_DEVICE_STATUS_TOPIC,
        bootstrap_servers=KAFKA_HOST,
        auto_offset_reset='latest',
        enable_auto_commit=False
    )
    # 将消费者的偏移量设置为最后一条消息的偏移量
    """
            同步发送 数据
            :param data_li:  发送数据
            :return:
            """
    while True:
        time.sleep(1)
        for message in consumer:
            try:
                if message:
                    message_key = message.key.decode('latin1')
                    message_value = message.value.decode('latin1')
                    vir_gantry_light = json.loads(message_value)
                    print("kafka消费者接收到主路限速和匝道管控信号", message_key, message_value)
                    if 'taskId' in vir_gantry_light:
                        if vir_gantry_light['taskId'] == key:
                            if 'frameStrategies' in vir_gantry_light:
                                vir_gantry_frame = vir_gantry_light['frameStrategies']
                                for vir_gantry in vir_gantry_frame:
                                    gantryId = vir_gantry['frameId']
                                    speedLimitValue = vir_gantry['speedLimitValue']
                                    emergencyLight = vir_gantry['emergencyLight']
                                    specialLight = vir_gantry['specialLight']
                                    emergency_json = {
                                        "emergencyLight": int(emergencyLight)
                                    }
                                    special_json = {
                                        "specialLight": int(specialLight)
                                    }
                                    SPEED_LIMIT_SIGN = True
                                    for i in range(len(sumo_main.speed_gantry_asc)):
                                        if gantryId in sumo_main.speed_gantry_asc[i]:
                                            index = sumo_main.speed_gantry_asc[i].index(gantryId)
                                            sumo_main.speed_limit_list[i][index] = int(speedLimitValue)
                                    sumo_main.emergency_dict[gantryId] = emergency_json
                                    sumo_main.special_dict[gantryId] = special_json
                                    # sumo_main.emergency_dict=emergency_dict
                                    # sumo_main.special_dict=special_dict
                        if 'rampStrategy' in vir_gantry_light:
                            rampStrategy = vir_gantry_light['rampStrategy']
                            if 'rampLight' in rampStrategy:
                                vir_ramp_light = int(rampStrategy['rampLight'])
                                sumo_main.vir_ramp_light = vir_ramp_light
            except Exception as e:
                print(str(e))
def MTF_receive_message(userId, task_id, resource):
    print('kafka 消费被触发')
    # 首先定义全局变量，用于在不同方法内互相通信
    global vir_gantry_dict, vir_ramp_light, SPEED_LIMIT_SIGN, \
        speed_limit_point_list, speed_limit_list, speed_gantry_asc, \
        emergency_dict, special_dict
    key = None
    if resource == 0:
        # key = str(userId) + '_' + str(task_id) + '_' + str(resource)
        key = str(task_id)
    elif resource == 1:
        key = str(task_id)

    # key = str(579) + '_' + str(224) + '_' + str(1)
    consumer = KafkaConsumer(
        MTF_VIR_DEVICE_STATUS_TOPIC,
        bootstrap_servers=KAFKA_HOST,
        auto_offset_reset='latest',
        enable_auto_commit=False
    )
    # 将消费者的偏移量设置为最后一条消息的偏移量
    """
            同步发送 数据
            :param data_li:  发送数据
            :return:
            """
    while True:
        time.sleep(1)
        for message in consumer:
            try:
                if message:
                    message_key = message.key.decode('latin1')
                    message_value = message.value.decode('latin1')
                    vir_gantry_light = json.loads(message_value)
                    print("kafka消费者接收到主路限速和匝道管控信号", message_key, message_value)
                    if 'taskId' in vir_gantry_light:
                        if vir_gantry_light['taskId'] == key:
                            if 'frameStrategies' in vir_gantry_light:
                                vir_gantry_frame = vir_gantry_light['frameStrategies']
                                for vir_gantry in vir_gantry_frame:
                                    gantryId = vir_gantry['frameId']
                                    speedLimitValue = vir_gantry['speedLimitValue']
                                    # emergencyLight = vir_gantry['emergencyLight']
                                    # specialLight = vir_gantry['specialLight']
                                    # emergency_json = {
                                    #     "emergencyLight": int(emergencyLight)
                                    # }
                                    # special_json = {
                                    #     "specialLight": int(specialLight)
                                    # }
                                    sumo_main.SPEED_LIMIT_SIGN = True
                                    for i in range(len(sumo_main.speed_gantry_asc)):
                                        if gantryId in sumo_main.speed_gantry_asc[i]:
                                            index = sumo_main.speed_gantry_asc[i].index(gantryId)
                                            sumo_main.speed_limit_list[i][index] = int(speedLimitValue)
                                    # sumo_main.emergency_dict[gantryId] = emergency_json
                                    # sumo_main.special_dict[gantryId] = special_json
                                    # sumo_main.emergency_dict=emergency_dict
                                    # sumo_main.special_dict=special_dict
                        if 'rampStrategy' in vir_gantry_light:
                            rampStrategy = vir_gantry_light['rampStrategy']
                            if 'rampLight' in rampStrategy:
                                if rampStrategy['rampId'] == "ramp-02":
                                    vir_ramp_light = int(rampStrategy['rampLight'])
                                    sumo_main.vir_ramp_light = vir_ramp_light
            except Exception as e:
                print(str(e))

# 接收自动驾驶车辆的实时数据
def receive_message_of_AV_vehicle(userId, task_id, resource):
    print("接收自动驾驶车辆的实时数据", 'kafka 消费被触发')
    global real_vehicle_queue
    # 首先定义全局变量，用于在不同方法内互相通信
    consumer = KafkaConsumer(
        AV_VEHICLE_TOPIC,
        bootstrap_servers=KAFKA_HOST,
        auto_offset_reset='latest',
        enable_auto_commit=False
    )
    # 将消费者的偏移量设置为最后一条消息的偏移量
    """
            同步发送 数据
            :param data_li:  发送数据
            :return:
            """
    while True:
        time.sleep(0.1 / 6)
        for message in consumer:
            if message:
                try:
                    message_key = message.key.decode('utf-8')
                    message_value = message.value.decode('utf-8')
                    vehicle_info = json.loads(message_value)
                    # print("kafka消费者接收到的真实车辆数据", message_key, message_value)
                    """
                    - 人驾车辆：Human-Driven Vehicle (HDV)
    
                    - 自动驾驶车辆：Autonomous Vehicle (AV)
                    
                    - 网联车辆：Connected Vehicle (CV)
                    """
                    UEID = vehicle_info["UEID"]
                    nSelfLat = vehicle_info["nSelfLat"]
                    nSelfLong = vehicle_info["nSelfLong"]
                    nSelfTrack = vehicle_info["nSelfTrack"]
                    nSelfSpeed = vehicle_info["nSelfSpeed"]
                    if real_vehicle_queue.qsize() < 2000:

                        real_vehicle_queue.put({"id": UEID, "lo": nSelfLong, "la": nSelfLat, "speed": nSelfSpeed,
                                                "ag": nSelfTrack}, block=False)
                except Exception as e:
                    print(str(e))
